package com.google.code.coma.assembly;

import com.google.code.coma.Coma;
import com.google.code.coma.domain.Computer;
import com.google.code.coma.domain.Order;
import com.google.code.coma.domain.Part;
import com.google.code.coma.space.ComputerPartSelectorCallback;
import com.google.code.coma.space.SelectAllSelectorCallback;
import com.google.code.coma.space.SpaceSelector;
import com.google.code.coma.space.util.SpaceManager;
import com.google.code.coma.worker.Worker;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.mozartspaces.core.*;
import org.mozartspaces.notifications.Notification;
import org.mozartspaces.notifications.NotificationListener;
import org.mozartspaces.notifications.NotificationManager;
import org.mozartspaces.notifications.Operation;

import java.io.Serializable;
import java.net.URI;
import java.util.*;

/**
 * @author Dominik Strasser, dominikstr@gmail.com
 */
public class SpaceAssemblyWorker extends AbstractAssemblyWorker implements Worker {
	private static final Log log = LogFactory.getLog(SpaceAssemblyWorker.class);

	private MzsCore core;

	private Capi capi;

	private ContainerReference parts;

	private ContainerReference orders;

	private ContainerReference computers;

	private Notification ordersNotification;

	private Notification partsNotification;

	private URI spaceUri;

	public SpaceAssemblyWorker(String factoryId) {
		super(factoryId);
		core = DefaultMzsCore.newInstance(0);
		capi = new Capi(core);
		spaceUri = URI.create("xvsm://localhost:4242");
		try {
			parts = SpaceManager.getOrCreateNamedContainer(spaceUri, factoryId + "_" + Coma.PARTS_CONTAINER, capi);
			computers = SpaceManager.getOrCreateNamedContainer(spaceUri, factoryId + "_" + Coma.COMPUTER_CONTAINER, capi);
			orders = SpaceManager.getOrCreateNamedContainer(spaceUri, factoryId + "_" + Coma.ORDER_CONTAINER, capi);
		} catch (MzsCoreException e) {
			core.shutdown(false);
			throw new RuntimeException("Unable to get container", e);
		}
	}

	@Override
	public void startWork() {
		NotificationManager notificationManager = new NotificationManager(core);
		try {
			ordersNotification = notificationManager.createNotification(orders, new NotificationListener() {
				@Override
				public void entryOperationFinished(Notification source, Operation operation, List<? extends Serializable> entries) {
					checkForComputerOrder();
				}
			}, Operation.WRITE);
			partsNotification = notificationManager.createNotification(parts, new NotificationListener() {
				@Override
				public void entryOperationFinished(Notification source, Operation operation, List<? extends Serializable> entries) {
					checkForComputerOrder();
					doAssembly(null);
				}
			}, Operation.WRITE);
		} catch (MzsCoreException e) {
			log.error("Space error", e);
		} catch (InterruptedException e) {
			log.error("Interrupted");
		}
		if (!checkForComputerOrder()) {
			while (doAssembly(null)) ;
		}
	}

	private boolean checkForComputerOrder() {
		boolean okay = false;
		try {
			List<Order> retrievedOrders = capi.read(orders, new SpaceSelector(new SelectAllSelectorCallback()), MzsConstants.RequestTimeout.DEFAULT, null);
			for (Order order : retrievedOrders) {
				if (order.isFinished()) continue; //skip finished orders
				if (!okay)
					okay = doAssembly(order);
			}

		} catch (MzsCoreException e) {
			log.error("Space error", e);
		}
		return okay;
	}

	private boolean doAssembly(Order order) {
		TransactionReference tx;
		try {
			tx = capi.createTransaction(MzsConstants.TransactionTimeout.INFINITE, spaceUri);
		} catch (MzsCoreException e) {
			log.error("Unable to create transaction", e);
			return false;
		}
		try {
			List<Part> list = capi.take(parts, new SpaceSelector(new ComputerPartSelectorCallback(order)), MzsConstants.RequestTimeout.INFINITE, tx);
			if (list == null || list.isEmpty()) return false;
			if (log.isInfoEnabled()) {
				log.info("Creating computer " + (order != null ? "for order " + order.getId() : "") + " out of " + list);
			}
			Computer computer = new Computer(getWorkerId(), list);
			computer.setOrder(order);
			capi.write(new Entry(computer), computers, MzsConstants.RequestTimeout.INFINITE, tx);
			capi.commitTransaction(tx);
			return true;
		} catch (MzsCoreException e) {
			log.error("Space exception", e);
			try {
				capi.rollbackTransaction(tx);
			} catch (MzsCoreException e1) {
				log.error("Error in transaction rollback", e);
			}
		}
		return false;
	}

	@Override
	public void stopWork() {
		if (ordersNotification != null) {
			try {
				ordersNotification.destroy();
			} catch (MzsCoreException e) {
				log.error(e);
			}
		}
		if (partsNotification != null) {
			try {
				partsNotification.destroy();
			} catch (MzsCoreException e) {
				log.error(e);
			}
		}
	}

	@Override
	public boolean isActive() {
		return !(ordersNotification == null || partsNotification == null) && (ordersNotification.isActive() || partsNotification.isActive());
	}
}
